package com.study.flink.datastream

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow

/**
  * CountWindow
  *
  * @author stephen
  * @date 2019-07-19 17:41
  */
object FlinkCountWindowDemo {

  def main(args: Array[String]): Unit = {
    // 初始化 Flink 执行环境
    val executionEnvironment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 导入隐式转换
    import org.apache.flink.api.scala._
    val socketDStream: DataStream[String] = executionEnvironment.socketTextStream("localhost", 1234)
    val mapDStream: DataStream[(String, Int)] = socketDStream.map(e => {
      val strings: Array[String] = e.split(" ")
      (strings(0), strings(1).toInt)
    })
    val keyDStream: KeyedStream[(String, Int), Tuple] = mapDStream.keyBy(0)
    // 滚动窗口 只有等相同key的元素个数达到3的时候才会进行 reduce 和 print 操作
    //val windowDStream: WindowedStream[(String, Int), Tuple, GlobalWindow] = keyDStream.countWindow(3)
    // 滑动窗口 只有等相同key的元素个数达到2的时候才会对该 key 的前4条数据进行 reduce 和 print 操作
    val windowDStream: WindowedStream[(String, Int), Tuple, GlobalWindow] = keyDStream.countWindow(4, 2)
    val reduceDStream: DataStream[(String, Int)] = windowDStream.reduce((e1, e2) => (e1._1, e1._2 + e2._2))
    reduceDStream.print()

    // 启动 Flink 应用
    executionEnvironment.execute("Flink CountWindow Demo")
  }
}
